package com.hery.functions;

import com.hery.pojo.Rule;
import com.hery.services.RulesService;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.time.LocalDateTime;
import java.util.Map;


@Slf4j
public class CustomMysqlSourceFunction extends RichSourceFunction<Rule> {

    private RulesService rulesService;

    private boolean flag = true;


    public CustomMysqlSourceFunction(Configuration parameters) {

        rulesService = new RulesService(parameters);
    }


    @Override
    public void run(SourceContext<Rule> ctx) throws InterruptedException {
        while (flag) {
            if (true || LocalDateTime.now().getSecond() == 0) {
                log.debug("rulesService all Sync Rule rules");
                rulesService.allSyncRule();
                for (Map.Entry<Integer, Rule> entry : rulesService.getRules().entrySet()) {
                    if (entry.getValue().isValid()) {
                        ctx.collect(entry.getValue());
                    }
                }
            }
            Thread.sleep(1000 * 60);
        }
        log.info("CustomMysqlSourceFunction FINISHED, flag is :{}", flag);
    }

    @Override
    public void cancel() {
        flag = false;
    }
}
